package cn.csg.ams.collect2.poptopics.datahub;

import cn.hutool.core.util.IdUtil;
import com.aliyun.datahub.client.model.Field;
import com.aliyun.datahub.client.model.FieldType;
import com.aliyun.datahub.client.model.RecordSchema;
import com.aliyun.datahub.client.model.TupleRecordData;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class TupleRecordDataUtils {

    /**
     * 获取String类型字段
     *
     * @param data  数据
     * @param index 索引
     * @return 字段
     */
    public static String getStringField(TupleRecordData data, int index) {
        Object field = data.getField(index);
        if (field == null) {
            return null;
        }
        return field.toString();
    }

    /**
     * 获取String类型字段
     *
     * @param data     数据
     * @param fileName 索引
     * @return 字段
     */
    public static String getStringField(TupleRecordData data, String fileName) {
        Object field = data.getField(fileName);
        if (field == null) {
            return null;
        }
        return field.toString();
    }

    /**
     * 获取String类型字段
     *
     * @param data  数据
     * @param index 索引
     * @return 字段
     */
    public static Long getLongField(TupleRecordData data, int index) {
        Object field = data.getField(index);
        if (field == null) {
            return null;
        }
        return Long.valueOf(field.toString());
    }

    /**
     * 获取String类型字段
     *
     * @param data     数据
     * @param fileName 字段名称
     * @return 字段
     */
    public static Long getLongField(TupleRecordData data, String fileName) {
        Object field = data.getField(fileName);
        if (field == null) {
            return null;
        }
        return Long.valueOf(field.toString());
    }


    /**
     * 获取BigDecimal类型字段
     *
     * @param data  数据
     * @param index 索引
     * @return 字段
     */
    public static BigDecimal getBigDecimalField(TupleRecordData data, int index) {
        Object field = data.getField(index);
        if (field == null) {
            return null;
        }
        return field instanceof BigDecimal ? (BigDecimal) field : new BigDecimal(field.toString());
    }

    /**
     * 获取BigDecimal类型字段
     *
     * @param data      数据
     * @param filedName 字段名称
     * @return 字段
     */
    public static BigDecimal getBigDecimalField(TupleRecordData data, String filedName) {
        Object field = data.getField(filedName);
        if (field == null) {
            return null;
        }
        return field instanceof BigDecimal ? (BigDecimal) field : new BigDecimal(field.toString());
    }


    /**
     * 创建Schema结构
     */
    public static RecordSchema getRecordSchema() {
        RecordSchema schema = new RecordSchema();
        schema.addField(new Field("alarm_id", FieldType.STRING));
        schema.addField(new Field("run_term_id", FieldType.STRING));
        schema.addField(new Field("run_meter_id", FieldType.STRING));
        schema.addField(new Field("measure_point_no", FieldType.BIGINT));
        schema.addField(new Field("alarm_code", FieldType.STRING));
        schema.addField(new Field("alarm_ext_data", FieldType.STRING));
        schema.addField(new Field("is_fals", FieldType.STRING));
        schema.addField(new Field("is_rest", FieldType.STRING));
        schema.addField(new Field("alarm_name", FieldType.STRING));
        schema.addField(new Field("alarm_level_code", FieldType.STRING));
        schema.addField(new Field("alarm_type_code", FieldType.STRING));
        schema.addField(new Field("alarm_sour_code", FieldType.STRING));
        schema.addField(new Field("alarm_class", FieldType.STRING));
        schema.addField(new Field("alarm_creat_way", FieldType.STRING));
        schema.addField(new Field("is_disp", FieldType.STRING));
        schema.addField(new Field("proc_pers", FieldType.STRING));
        schema.addField(new Field("terminal_type", FieldType.STRING));
        schema.addField(new Field("work_order_mode", FieldType.STRING));
        schema.addField(new Field("order_id", FieldType.STRING));
        schema.addField(new Field("order_state", FieldType.STRING));
        schema.addField(new Field("statute_alarm_code", FieldType.STRING));
        schema.addField(new Field("area_code", FieldType.STRING));
        schema.addField(new Field("pre_rece_time", FieldType.STRING));
        schema.addField(new Field("alarm_time", FieldType.STRING));
        schema.addField(new Field("rest_time", FieldType.STRING));
        return schema;
    }


    /**
     * 创建Schema结构 (topic_ods_0300domk_pr_pf_outage_recovery_mon_r)
     */
    public static RecordSchema getRecordSchema2() {
        RecordSchema schema = new RecordSchema();
        schema.addField(new Field("manage_org_no_province", FieldType.STRING));
        schema.addField(new Field("manage_org_no_city", FieldType.STRING));
        schema.addField(new Field("manage_org_no_area", FieldType.STRING));
        schema.addField(new Field("supply_org_no", FieldType.STRING));
        schema.addField(new Field("area_code", FieldType.STRING));
        schema.addField(new Field("sub_no", FieldType.STRING));
        schema.addField(new Field("line_no", FieldType.STRING));
        schema.addField(new Field("ta_no", FieldType.STRING));
        schema.addField(new Field("low_branch_no", FieldType.STRING));
        schema.addField(new Field("outage_event_id", FieldType.STRING));
        schema.addField(new Field("obj_id", FieldType.STRING));
        schema.addField(new Field("obj_name", FieldType.STRING));
        schema.addField(new Field("obj_type", FieldType.STRING));
        schema.addField(new Field("outage_status", FieldType.STRING));
        schema.addField(new Field("outage_start_time", FieldType.STRING));
        schema.addField(new Field("outage_end_time", FieldType.STRING));
        schema.addField(new Field("outage_duration", FieldType.BIGINT));
        schema.addField(new Field("alarm_type", FieldType.STRING));
        schema.addField(new Field("judge_rule", FieldType.STRING));
        schema.addField(new Field("run_meter_id", FieldType.STRING));
        schema.addField(new Field("run_term_id", FieldType.STRING));
        schema.addField(new Field("term_type_code", FieldType.STRING));
        schema.addField(new Field("gisId", FieldType.STRING));
        return schema;
    }

    /**
     * 创建Schema结构 (topic_ods_0300domk_alarm_event_r)
     */
    public static RecordSchema getAlarmEventRecordSchema() {
        RecordSchema schema = new RecordSchema();
        List<Field> fields = getAlarmEventRecordSchemaFields();
        for (Field field : fields) {
            schema.addField(field);
        }
        return schema;
    }


    public static List<Field> getAlarmEventRecordSchemaFields() {
        List<Field> fields = new ArrayList<>();
        fields.add(new Field("alarm_id", FieldType.STRING));
        fields.add(new Field("run_term_id", FieldType.STRING));
        fields.add(new Field("run_meter_id", FieldType.STRING));
        fields.add(new Field("measure_point_no", FieldType.BIGINT));
        fields.add(new Field("alarm_code", FieldType.STRING));
        fields.add(new Field("alarm_ext_data", FieldType.STRING));
        fields.add(new Field("is_fals", FieldType.STRING));
        fields.add(new Field("is_rest", FieldType.STRING));
        fields.add(new Field("alarm_name", FieldType.STRING));
        fields.add(new Field("alarm_level_code", FieldType.STRING));
        fields.add(new Field("alarm_type_code", FieldType.STRING));
        fields.add(new Field("alarm_sour_code", FieldType.STRING));
        fields.add(new Field("alarm_class", FieldType.STRING));
        fields.add(new Field("alarm_creat_way", FieldType.STRING));
        fields.add(new Field("terminal_type", FieldType.STRING));
        fields.add(new Field("status_alarm_code", FieldType.STRING));
        fields.add(new Field("area_code", FieldType.STRING));
        fields.add(new Field("pre_rece_time", FieldType.STRING));
        fields.add(new Field("alarm_time", FieldType.STRING));
        fields.add(new Field("rest_time", FieldType.STRING));
        return fields;
    }


    public static TupleRecordData getCurAndVlotSchema() {
        RecordSchema schema = new RecordSchema();
        schema.addField(new Field("run_meter_id", FieldType.STRING));
        schema.addField(new Field("data_time", FieldType.TIMESTAMP));
        schema.addField(new Field("preposition_time", FieldType.TIMESTAMP));
        schema.addField(new Field("p_volt_a", FieldType.DECIMAL));
        schema.addField(new Field("p_volt_b", FieldType.DECIMAL));
        schema.addField(new Field("p_volt_c", FieldType.DECIMAL));
        schema.addField(new Field("p_curr_a", FieldType.DECIMAL));
        schema.addField(new Field("p_curr_b", FieldType.DECIMAL));
        schema.addField(new Field("p_curr_c", FieldType.DECIMAL));
        schema.addField(new Field("zl_curr", FieldType.DECIMAL));
        schema.addField(new Field("data_source_code", FieldType.STRING));
        schema.addField(new Field("area_code", FieldType.STRING));
        schema.addField(new Field("load_time", FieldType.TIMESTAMP));

        // 模拟数据
        TupleRecordData recordData = new TupleRecordData(schema);
        recordData.setField("run_meter_id", "JL1100867");
        recordData.setField("data_time", 1711707300000000L);
        // recordData.setField("preposition_time", FieldType.TIMESTAMP);
        recordData.setField("p_volt_a", new BigDecimal("235.4"));
        recordData.setField("p_volt_b", new BigDecimal("235.4"));
        recordData.setField("p_volt_c", new BigDecimal("235.4"));
        recordData.setField("p_curr_a", new BigDecimal("235.4"));
        recordData.setField("p_curr_b", new BigDecimal("235.4"));
        recordData.setField("p_curr_c", new BigDecimal("235.4"));
        recordData.setField("zl_curr", new BigDecimal("235.4"));
        // recordData.setField("data_source_code", FieldType.STRING);
        recordData.setField("area_code", "032000");
        // recordData.setField("load_time", FieldType.TIMESTAMP);

        return recordData;
    }

    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 将kafka中的数据转换为datahub中的数据
     */
    public static TupleRecordData kafkaToDatahubValue(String value) {
        try {
            HashMap hashMap = objectMapper.readValue(value, HashMap.class);

            // 创建TupleRecordData
            RecordSchema schema = getRecordSchema();
            TupleRecordData recordData = new TupleRecordData(schema);
            for (Field field : schema.getFields()) {
                String fieldName = field.getName();
                recordData.setField(fieldName, hashMap.get(fieldName));
            }
            recordData.setField("alarm_id", IdUtil.getSnowflake().nextIdStr());
            return recordData;
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
}
